package procedure0;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;


@SuppressWarnings("deprecation")
public class MapReduceTest extends Configured implements Tool{

	@Override
	public int run(String[] args) throws Exception {
		// TODO Auto-generated method stub
		if (args.length != 2) {
			System.err.println("Usage: parameter <in> <out>");
			System.exit(2);
		}
		
		Configuration globalConf = getConf();
		globalConf.set("eigenvaluePath", args[1]);
		String tableNames = "point";
//		String tableNames = "";
		createSpecialPath(globalConf, args, tableNames);
		globalConf.set("latitude", "double");
		globalConf.set("id", "String");
		globalConf.set("asdiflightplan_id", "String");
		globalConf.set("ordinal", "double");
		globalConf.set("longitude", "double");


		FileSystem fs = FileSystem.get(globalConf);
		fs.rename(new Path(args[1] + "/temp%_%point%_%0"), new Path(args[1] + "/MatchResult"));
		globalConf.set("mapred.textoutputformat.separator", "\t");
		Job job0 = new Job(globalConf, "calculate result limitation");
		job0.setJarByClass(MapReduceTest.class);
		job0.setMapperClass(MR_ResultLimitMapper.class);
		job0.setOutputKeyClass(Text.class);
		job0.setOutputValueClass(Text.class);
		FileInputFormat.addInputPath(job0, new Path(args[1] + "/MatchResult"));
		FileOutputFormat.setOutputPath(job0, new Path(args[1] + "/Result"));
		job0.submit();
		job0.waitForCompletion(true);


		DelFolder(args[1], "Result");
		return 0;
	}
	
	private void DelFolder(String delFolder, String remainFolder) {

		File file = new File(delFolder);
		if (!file.exists() || !file.isDirectory()) {
			return;
		}

		String[] tempFiles = file.list();
		File temp = null;
		for (int i = 0; i < tempFiles.length; i++) {
			if (tempFiles[i].equals(remainFolder))
				continue;
			if (delFolder.endsWith(File.separator)) {
				temp = new File(delFolder + tempFiles[i]);
			} else {
				temp = new File(delFolder + File.separator + tempFiles[i]);
			}
			if (temp.isFile()) {
				temp.delete();
			}
			if (temp.isDirectory()) {
				DelFolder(temp.getPath(), "");// 先递归删除文件夹里面的文件
				temp.delete();// 再删除空文件夹
			}
		}
	}

	private void createSpecialPath(Configuration globalConf, String[] args, String tableNames) throws IOException {
		// TODO Auto-generated method stub
		FileSystem fs = FileSystem.get(globalConf);
		String[] tns = tableNames.split(" ");
		for (int i=0; i<tns.length; i++) {
			Path path = new Path(args[0] + "/" + tns[i]);
			Path targetPath = new Path(args[1] + "/temp%_%" + tns[i] + "%_%0/" + tns[i]);
			FSDataInputStream fin = fs.open(path);
			BufferedReader br = new BufferedReader(new InputStreamReader(fin, "UTF-8"));
			FSDataOutputStream fout = fs.create(targetPath);
			BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(fout, "UTF-8"));
			
			String line;
			while ((line=br.readLine()) != null) {
				bw.write(line + "\n");
				bw.flush();
			}
			
			bw.close();
			br.close();
		}
	}

	public static void main(String[] args) throws Exception {
		// TODO Auto-generated method stub
		int res = ToolRunner.run(new Configuration(), new MapReduceTest(), args);
		System.exit(res);
	}
}
